//package com.huatai.bi.kafaka;
//
//import cn.hutool.core.bean.BeanUtil;
//import cn.hutool.core.util.ObjectUtil;
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.baomidou.mybatisplus.core.toolkit.Wrappers;
//import com.huatai.bi.constant.KafkaConstant;
//import com.huatai.bi.entity.*;
//import com.huatai.bi.handler.dataSet.PythonCode;
//import com.huatai.bi.kafaka.DTO.DataSetKafkaDTO;
//import com.huatai.bi.service.*;
//import com.huatai.bi.service.impl.ClickHouseJDBCService;
//import com.huatai.common.error.exception.ServiceException;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.stereotype.Component;
//
//import java.util.Date;
//import java.util.List;
//
///**
// * @BelongsProject: itc-airport-cloud
// * @BelongsPackage: com.huatai.bi.kafaka
// * @Author: hecaiy
// * @CreateTime: 2024-07-04  10:43
// * @Description: TODO python数据集创建操作
// * @Version: 1.0
// */
//@Component
//@Slf4j
//public class DataSetPythonKafkaListener {
//    @Autowired
//    private DirectoryService directoryService;
//    @Autowired
//    private ClickHouseJDBCService clickHouseJDBCService;
//    @Autowired
//    private DataSetColumnService dataSetColumnService;
//    @Autowired
//    private DirectDataSetService directDataSetService;
//    @Autowired
//    private PythonCode pythonCode;
//    @Autowired
//    private XxlJobInfoService xxlJobInfoService;
//    @Autowired
//    private DataSynrecordService dataSynrecordService;
//
//    /**
//     * 1.python构建数据集
//     * 2.创建主附表
//     * 3.数据同步
//     * 4.xxl-job任务调度
//     * @param record
//     */
//    @KafkaListener(groupId = "bi_data_set_python",topics = KafkaConstant.KAFKA_TOPIC_DATA_SET_PYTHON)
//    public void dataSetListener(ConsumerRecord record){
//        String message = (String) record.value();
//        DataSetKafkaDTO directoryParam = JSON.parseObject(message, DataSetKafkaDTO.class);
//        Long dtId = directoryParam.getDtId();
//        DirectoryEntity directoryEntity = directoryService.getById(dtId);
//        if(ObjectUtil.isEmpty(directoryEntity)){
//            log.error("数据集信息为空");
//            throw new ServiceException("数据集信息为空");
//        }
//        if(ObjectUtil.isNotEmpty(directoryEntity.getVersion()) && ObjectUtil.isNotEmpty(directoryEntity.getTableName())){
//            log.error("数据集已存在");
//            throw new ServiceException("数据集已存在");
//        }
//        directoryEntity.setTableName(KafkaConstant.DATA_SET_TYPE_PYTHON_MASTER+dtId);
//        directoryEntity.setVersion(KafkaConstant.DATA_SET_TYPE_PYTHON_SLAVE+dtId);
//        //数仓数据表创建，数据同步
//        List<DataSetColumnEntity> list = dataSetColumnService.list(Wrappers.<DataSetColumnEntity>lambdaQuery().eq(DataSetColumnEntity::getDataSetId, dtId)
//			.orderByAsc(DataSetColumnEntity::getColumnPosition));
//        if(ObjectUtil.isEmpty(list)){
//            log.error("数据集字段为空");
//            throw new ServiceException("数据集字段为空");
//        }
//        clickHouseJDBCService.createTableByColumnList(directoryEntity.getTableName(), list,"");
//        clickHouseJDBCService.createTableByColumnList(directoryEntity.getVersion(),list,"");
//        DirectDataSetEntity dataSetEntity = directDataSetService.getById(dtId);
//        List<Object> objects = pythonCode.previewData(dataSetEntity,null,null);
//        if(ObjectUtil.isNotEmpty(objects)){
//            clickHouseJDBCService.insertDataByTableNameObject(directoryEntity.getTableName(),objects);
//        }
//        this.directoryService.updateById(directoryEntity);
//		//TODO  此处差创建xxl调度任务
//		if(StringUtils.isEmpty(directoryParam.getCron())){
//			return;
//		}
//        XxlJobInfoEntity xxlJobInfoEntity = new XxlJobInfoEntity();
//        xxlJobInfoEntity.setJobGroup(4);
//        xxlJobInfoEntity.setJobDesc("数据集SQL调度任务"+directoryEntity.getName());
//        xxlJobInfoEntity.setAddTime(new Date());
//        xxlJobInfoEntity.setUpdateTime(new Date());
//        xxlJobInfoEntity.setAuthor("lowcode_bi");
//        xxlJobInfoEntity.setAlarmEmail("waytale@waytale.cn");
//        xxlJobInfoEntity.setScheduleType("CRON");
//        xxlJobInfoEntity.setScheduleConf("*/5 * * * * ");
//        xxlJobInfoEntity.setMisfireStrategy("DO_NOTHING");
//        xxlJobInfoEntity.setExecutorRouteStrategy("FIRST");
//        xxlJobInfoEntity.setExecutorHandler("PythonHandler");
//        JSONObject executorParam = new JSONObject();
//        executorParam.put("dtId",dtId);
//        xxlJobInfoEntity.setExecutorParam(JSON.toJSONString(executorParam));
//        xxlJobInfoEntity.setExecutorFailRetryCount(0);
//        xxlJobInfoEntity.setGlueType("BEAN");
//        xxlJobInfoEntity.setExecutorBlockStrategy("SERIAL_EXECUTION");
//        xxlJobInfoEntity.setExecutorTimeout(0);
//        xxlJobInfoService.saveOrUpdateXxlJob(xxlJobInfoEntity);
//
//        //记录创建的调度任务
//        DataSynrecordEntity entity = dataSynrecordService.getOne(Wrappers.<DataSynrecordEntity>lambdaQuery()
//                .eq(DataSynrecordEntity::getDtId, dtId));
//        DataSynrecordEntity dataSynrecordEntity = new DataSynrecordEntity();
//        //若存在调度任务 停止原来的调度任务、删除原来的调度任务
//        if(ObjectUtil.isNotEmpty(entity)){
//            BeanUtil.copyProperties(entity,dataSynrecordEntity);
//            xxlJobInfoService.stop(entity.getTaskId());
//            xxlJobInfoService.removeJob(entity.getTaskId());
//        }
//        dataSynrecordEntity.setDtId(dtId);
//        dataSynrecordEntity.setTableId(dtId);
//        dataSynrecordEntity.setTaskId(xxlJobInfoEntity.getId());
//        dataSynrecordEntity.setId(dataSynrecordEntity.getId());
//        dataSynrecordEntity.setDirId(directoryEntity.getPid());
//        dataSynrecordEntity.setTargetTableName(directoryEntity.getTableName());
//        dataSynrecordEntity.setCron(directoryParam.getCron());
//        dataSynrecordEntity.setIsExec(1);
//        dataSynrecordEntity.setSynType(1);
//        dataSynrecordService.saveOrUpdate(dataSynrecordEntity);
//
//        //启动执行
//        xxlJobInfoService.start(xxlJobInfoEntity.getId());
//    }
//}
